package com.bytes32.classification.tree.dataset;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.nio.DoubleBuffer;
import java.text.DecimalFormat;
import java.text.NumberFormat;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Iterator;

import org.apache.hadoop.io.WritableUtils;

import com.bytes32.stats.Config;
import com.bytes32.stats.LargeQuantiles;
import com.bytes32.stats.QBuffer;
import com.bytes32.stats.ReduceQuantiles;

/**
 * Used for continuous values as well as dates 
 * @author murariuf
 *
 */
public final class OrderedProcessor implements FeatureProcessor<LargeQuantiles>{

	/**
	 * Un-sampled quantiles up to max 100
	 * TODO: maybe this can be configurable
	 */
	private LargeQuantiles quantiles = new LargeQuantiles(Integer.MAX_VALUE, Config._03);
	
	/* do some external buffering */
	private transient DoubleBuffer buffer = DoubleBuffer.allocate(Config._03.getK());

	/**
	 * Counts the values, this is useful because the quantiles
	 * returned split the domain into equal parts so if we know
	 * the total we can know how many elements are in every split
	 */
	private long totalCount = 0;
	
	@Override
	public void readFields(DataInput stream) throws IOException {
		this.totalCount = WritableUtils.readVLong(stream);
		this.quantiles = new LargeQuantiles(Integer.MAX_VALUE, Config._03);
		this.quantiles.readFields(stream);
	}

	@Override
	public void write(DataOutput stream) throws IOException {
		this.quantiles.add(this.buffer.array());
		this.buffer.clear();
		WritableUtils.writeVLong(stream, this.totalCount);
		this.quantiles.write(stream);
	}

	@Override
	public long getCount() {
		return this.quantiles.getK(); /* the maximum quantils we can get */
	}

	@Override
	public void handleString(String seq) throws InvalidFeatureValueException {
		/* attempt some simple decoding  first if it's a Long in hex value
		 * then a double in some form  */
		try {
			Long value = Long.decode(seq);
			bufferedAdd(value);
		} catch (NumberFormatException e) {
			/* FIXME: the Locale should be configurable you should be able to analyze a US dataset on a French box */
			NumberFormat format = DecimalFormat.getInstance(); 
			try {
				Double value = format.parse(seq.toString()).doubleValue();
				bufferedAdd(value);
			} catch (ParseException e1) {
				/* we have some garbage, no problem
				 * FIXME: add reporting about garbage */
			}
		}
	}
	
	private void bufferedAdd(double value){
		if (merged)
			throw new IllegalStateException(" Cannot add values while merging ");
		this.buffer.put(value);
		totalCount+=1;
		if (buffer.hasRemaining()){
			return;
		} else {
			this.quantiles.add(this.buffer.array());
			this.buffer.clear();
		}
	}

	@Override
	public Iterator<LargeQuantiles> iterator() {
		return new Iterator<LargeQuantiles>() {
			private boolean hasNext = true;
			public boolean hasNext() { return hasNext; }
			public LargeQuantiles next() { hasNext = false; return OrderedProcessor.this.quantiles; }
			public void remove() { throw new UnsupportedOperationException(" Operation not supported ");}
		};
	}
	
	/*
	 * Just a marker, not thread safe 
	 */
	private boolean merged = false;
	
	/**
	 * Merges another ordered processor
	 * into this one
	 * @param processor
	 */
	public OrderedProcessor merge(final Iterator<OrderedProcessor> buffer){
		merged = true;
		ReduceQuantiles result = new ReduceQuantiles(Config._03);
		this.quantiles =  result.reduce(new Iterator<LargeQuantiles>() {
			public boolean hasNext() { return buffer.hasNext(); }
			public LargeQuantiles next() {
				OrderedProcessor next = buffer.next();
				OrderedProcessor.this.totalCount+= next.totalCount;
				return next.quantiles;
			}
			public void remove() { throw new UnsupportedOperationException(" Operation not supported "); }
		});
		return this;
	}

	public long getTotalCount() {
		return totalCount;
	}

	@Override
	public String toString() {
		final int max = 25;
		ArrayList<Double> q = new ArrayList<Double>(); 
		for (double i = 1f/max; i < 1; i+=1f/max){
			q.add(i);
		}
		double[] quantiles = new double[q.size()];
		for (int i = 0; i < q.size(); i++){
			quantiles[i] = q.get(i);
		}
		return Arrays.toString(this.quantiles.quantile(quantiles));
	}
	
	@Override
	public void compact() {
		this.quantiles.compact();
	}
	
}
